package com.niodata.dp.plugin.core;

import java.io.File;
import java.io.IOException;
import java.security.PrivilegedAction;
import java.sql.SQLException;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

import com.niodata.dp.api.common.DpTaskApi;
import com.niodata.dp.collection.WrappedStringKeyMap;
import com.niodata.dt.fs.FileSystemBuilder;
import com.niodata.dt.fs.FileSystemType;
import javax.sql.DataSource;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.security.UserGroupInformation;

public class DefaultPluginContext implements PluginContext {

  String principal;
  Input config;
  String keyTabFile;
  String hadoopHome;
  String taskName;
  String jobName;
  String runTimeJobId;
  String flowExecId;
  String flowName;
  String configId;
  PluginService pluginService;
  FileSystem fileSystem;
  FileSystem tmpFileSystem;
  File workDir;
  UserGroupInformation userGroupInformation;
  Configuration hdfsConfiguration;
  boolean kerberosEnabled = false;
  private Map<String, DataSource> dataSourceMap = new HashMap<>();

  protected DpTaskApi taskApi;
  private Map<String, FileSystem> fsCache = new ConcurrentHashMap<>();

  public DefaultPluginContext(DpTaskApi taskApi) {
    this.taskApi = taskApi;
  }

  protected void init(Map<String, Object> confMap) throws IOException {

  }

  protected void checkFileSystem() throws IOException {
    if (this.userGroupInformation == null) {
      if (kerberosEnabled) {
        UserGroupInformation.setConfiguration(hdfsConfiguration);
        this.userGroupInformation = UserGroupInformation
              .loginUserFromKeytabAndReturnUGI(principal, keyTabFile);
        UserGroupInformation.setLoginUser(userGroupInformation);
      } else {
        this.userGroupInformation = UserGroupInformation.createRemoteUser(principal);
      }
    }
    if (fileSystem == null) {
      fileSystem = userGroupInformation.doAs(new PrivilegedAction<FileSystem>() {
        @Override
        public FileSystem run() {
          try {
            FileSystem fs = FileSystem
                  .get(hdfsConfiguration);
            return fs;
          } catch (IOException e) {
            e.printStackTrace();
          }
          return null;
        }
      });
    } else {
      userGroupInformation.checkTGTAndReloginFromKeytab();
    }
  }


  public String getPrincipal() {
    return this.principal;
  }

  public Input getConfig() {
    return this.config;
  }

  String getKeyTabFile() {
    return this.keyTabFile;
  }

  String getHadoopHome() {
    return this.hadoopHome;
  }

  public FileSystem getHadoopFileSystem() throws IOException {
    checkFileSystem();
    return this.fileSystem;
  }

  @Override
  public synchronized FileSystem getFileSystem(String dsName) throws IOException {
    if (dsName.equals("tmp")) {
      return getTmpFileSystem();
    }
    FileSystem fs = this.fsCache.get(dsName);
    if (fs == null) {
      WrappedStringKeyMap config = taskApi.getFsConfig(dsName, this.principal);
      FileSystemType fsType = FileSystemType.valueOf(config.getString("fsType"));
      Map<String, String> configMap = new HashMap<>();
      config.keySet().forEach(x->configMap.put(x, config.getString(x)));
      fs = FileSystemBuilder.createFileSystem(principal, fsType, configMap);
      fsCache.put(dsName, fs);
    }
    return fs;
  }

  @Override
  public FileSystem getTmpFileSystem() throws IOException {
    return this.tmpFileSystem;
  }

  @Override
  public File getWorkDir() {
    return this.workDir;
  }

  @Override
  public synchronized DataSource getDataSource(String dsName, String dbName) throws IOException, SQLException {
    String key = dsName + "—" + dbName;
    DataSource dataSource = dataSourceMap.get(key);
    if (dataSource == null) {
      dataSource = DataSourceUtil.newDataSource(taskApi, dsName, dbName, principal);
      dataSourceMap.put(key, dataSource);
    }
    return dataSource;
  }

  @Override
  public String getTaskName() {
    return this.taskName;
  }

  @Override
  public String getJobName() {
    return this.jobName;
  }

  @Override
  public String getRunTimeJobId() {
    return runTimeJobId;
  }

  @Override
  public String getFlowName() {
    return this.flowName;
  }

  @Override
  public String getFlowExecId() {
    return this.flowExecId;
  }

  @Override
  public String getScheduleId() {
    return null;
  }

  @Override
  public String getConfigId() {
    return this.configId;
  }

  @Override
  public void output(Output output) throws Exception {

  }



}